/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include <glog/logging.h>


#include "hadoop/Pipes.hh"
#include "hadoop/TemplateFactory.hh"
#include "hadoop/StringUtils.hh"
#include <QtCore/QCoreApplication>
#include <sstream>

const std::string NUMBERSUM = "NUMBERSUM";
const std::string INPUT_LINES = "INPUT_LINES";

class WordCountMap: public HadoopPipes::Mapper {
public:
  HadoopPipes::TaskContext::Counter* inputLines;
  
  WordCountMap(HadoopPipes::TaskContext& context) {
    inputLines = context.getCounter(NUMBERSUM, INPUT_LINES);
  }
  
  void map(HadoopPipes::MapContext& context) {
    QString qs=QString::fromStdString(context.getInputValue());
    bool ok;
    int value=qs.toInt(&ok,10);
    std::string mapvalue((char*)&value,sizeof(int));
    context.emitRecord("value", mapvalue);
    context.incrementCounter(inputLines, 1);
	LOG_EVERY_N(INFO, 10000) << "Got 10000 lines ";
  }
};

class WordCountReduce: public HadoopPipes::Reducer {
public:
  WordCountReduce(HadoopPipes::TaskContext& context) {

  }

  void reduce(HadoopPipes::ReduceContext& context) {
    int64_t sum = 0;
    while (context.nextValue()) {
      std::string v=context.getInputValue();
      int value;
      memcpy(&value,v.data(),sizeof(int));
      sum += value;
    }
    std::ostringstream oss;
    oss<<sum;
    std::string ret=oss.str();
    context.emitRecord(context.getInputKey(), ret);
  }
};

int main(int argc, char *argv[]) {
  setlocale(LC_ALL,"");
  google::InitGoogleLogging("numbersum");
//  google::InstallFailureSignalHandler();
  LOG(INFO) << "numbers sum started";
  QCoreApplication app(argc,argv);
  bool ret=HadoopPipes::runTask(HadoopPipes::TemplateFactory<WordCountMap, 
                              WordCountReduce>());

  app.quit();
  LOG(INFO) << "numbers sum end,ret="<<ret?"true":"false";
  return ret?0:-1;
}

